package com.gitee.dubboOps.config;

import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.config.annotation.Service;
import com.gitee.common.Util.EncryptyUtil;
import com.gitee.dubboOps.model.RedisServiceModel;
import com.gitee.dubboOps.model.ServiceDO;
import com.gitee.dubboOps.redis.redisImpl.RedisServiceImpl;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


@Service
@Configuration
public class ZkListener {
    @Value("${dubbo.registry.address}")
    private String zkUrl;

    private static final String ZK_PATH = "/dubbo";

    private static ConcurrentHashMap<String, ServiceDO> map = new ConcurrentHashMap<>();


    @Autowired
    RedisServiceImpl redisServiceimpl;


    public static List<ServiceDO> findProviderTargetsByDataId(CuratorFramework client, String DataId) throws Exception {
        String path = ZK_PATH + "/" + DataId + "/providers";
        List<ServiceDO> serviceDOS = new ArrayList<>();
        try {
            List<String> providers =  client.getChildren().forPath(path);
            for(String url: providers) {
                serviceDOS.add(getServiceDOFromProviderUrl(URL.valueOf(URL.decode(url))));
            }
        }catch (Exception e) {
            throw new Exception("findProviderTargetsByDataId,获取子节点出错"+ ExceptionUtils.getStackTrace(e));
        }

        return serviceDOS;
    }

    private List<URL> getPorviders(CuratorFramework client, String serviceName) throws Exception {
        List<URL> urls = new ArrayList<>();
        try {
            String path = serviceName + "/providers";
            List<String> providers = client.getChildren().forPath(path);
            for(String url: providers) {
                urls.add(URL.valueOf(URL.decode(url)));
            }
        } catch (Exception e) {
            throw new Exception("getPorviders,获取子节点出错"+ ExceptionUtils.getStackTrace(e));
        }
        return urls;
    }

    public static ServiceDO getServiceDOFromProviderUrl(URL url) {
        ServiceDO serviceDO = new ServiceDO();
        serviceDO.setOwner(url.getUsername());
        serviceDO.setIP(url.getIp());
        serviceDO.setPort(url.getPort());
        serviceDO.setTimeout(url.getParameter("default.timeout", 3000));
        serviceDO.setApplication(url.getParameter("application"));
        serviceDO.setServicename(url.getServiceInterface());
        serviceDO.setZone(url.getParameter("zone"));
        serviceDO.setGroup(url.getParameter("group"));
        serviceDO.setVersion(url.getParameter("version"));
        serviceDO.setCreate_time(new Date(url.getParameter("timestamp", 1547307164981L)));
        String sign = EncryptyUtil.md5( serviceDO.getIP() + serviceDO.getApplication() + serviceDO.getServicename()
        + serviceDO.getZone() + serviceDO.getGroup() + serviceDO.getVersion());
        serviceDO.setSgin(sign);

        return serviceDO;
    }


    private void addProviderEvent(String path) throws Exception{
        try {
            ServiceDO serviceDO = getServiceDOFromProviderUrl(URL.valueOf(URL.decode(path)));
            String sign = serviceDO.getSgin();
            if(!map.containsKey(sign)) {
                map.put(sign, serviceDO);
                RedisServiceModel redisServiceModel = new RedisServiceModel();
                redisServiceModel.setStringServiceDOMAP(map);
                redisServiceimpl.put(zkUrl, redisServiceModel, -1);
                System.out.println("addProviderEvent: add Provider==>" + path);
            }
        } catch (Exception e) {
            throw new Exception("addProviderEvent ERROR:" + ExceptionUtils.getStackTrace(e));
        }
    }

    private void removeProviderEvent(String path) throws Exception{
        try {
            ServiceDO serviceDO = getServiceDOFromProviderUrl(URL.valueOf(URL.decode(path)));
            String sign = serviceDO.getSgin();
            map.remove(sign);
            RedisServiceModel redisServiceModel = new RedisServiceModel();
            redisServiceModel.setStringServiceDOMAP(map);
            redisServiceimpl.put(zkUrl, redisServiceModel, -1);
        } catch (Exception e) {
            throw new Exception("removeProviderEvent ERROR:" + ExceptionUtils.getStackTrace(e));
        }
    }


    private void addListener(ExecutorService pool, CuratorFramework client, String path) throws Exception {
        final PathChildrenCache childrenCache = new PathChildrenCache(client, path, true);

        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        childrenCache.getListenable().addListener(
            new PathChildrenCacheListener() {

                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
                        throws Exception {
                    switch (event.getType()) {
                        case CHILD_ADDED:
                            addProviderEvent(event.getData().getPath());
                            break;
                        case CHILD_REMOVED:
                            removeProviderEvent(event.getData().getPath());
                            System.out.println("CHILD_REMOVED: " + event.getData().getPath());
                            break;
                        case CHILD_UPDATED:
                            System.out.println("CHILD_UPDATED: " + event.getData().getPath());
                            break;
                        default:
                            break;
                    }
                }
            },
            pool
        );

    }

    private void addServiceEvent(CuratorFramework client, String serviceName) throws Exception{
        try {
            List<URL> urls = getPorviders(client, serviceName);
            for(URL url: urls) {
                ServiceDO serviceDO = getServiceDOFromProviderUrl(url);
                String sign = serviceDO.getSgin();

                if(!map.containsKey(sign)) {
                    map.put(sign, serviceDO);
                    System.out.println("addServiceEvent: ADD Service==> " + serviceName + ", ZONE==>" + serviceDO.getZone());
                }
            }
            RedisServiceModel redisServiceModel = new RedisServiceModel();
            redisServiceModel.setStringServiceDOMAP(map);
            redisServiceimpl.put(zkUrl, redisServiceModel, -1);
        } catch (Exception e) {
            throw new Exception("addServiceEvent ERROR:"+ExceptionUtils.getStackTrace(e));
        }
    }

    private void removeService(CuratorFramework client, String serviceName, List<ServiceDO> serviceDOS) {
    }

    @Bean("client")
  //  @ConditionalOnProperty(name = "dubboops.zk.autoRegister", havingValue = "true")
    public  CuratorFramework getCuratorFramework() throws Exception{
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);

        CuratorFramework client = CuratorFrameworkFactory.newClient(zkUrl, retryPolicy);
        client.start();

        RedisServiceModel redisServiceModel = new RedisServiceModel();
        if(redisServiceimpl.isKeyExists(zkUrl)) {
            redisServiceModel = redisServiceimpl.get(zkUrl);

            if(redisServiceModel.getStringServiceDOMAP()!=null) {
                map = redisServiceModel.getStringServiceDOMAP();
            }
        }

        final ExecutorService pool = Executors.newFixedThreadPool(4);
        final PathChildrenCache childrenCache = new PathChildrenCache(client, ZK_PATH, true);
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        childrenCache.getListenable().addListener(
                new PathChildrenCacheListener() {

                    @Override
                    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
                            throws Exception {
                        switch (event.getType()) {
                            case CHILD_ADDED:
                                String serviceName = event.getData().getPath();
                                addServiceEvent(client, serviceName);

                                //监听provider地址列表
                                String path = serviceName + "/providers";
                                addListener(pool, client, path);

                                break;
                            case CHILD_REMOVED:
                                System.out.println("CHILD_REMOVED: " + event.getData().getPath());
                                break;
                            case CHILD_UPDATED:
                                System.out.println("CHILD_UPDATED: " + event.getData().getPath());
                                break;
                            default:
                                break;
                        }
                    }
                },
                pool
        );

        System.out.println("Register zk watcher successfully!");

        return client;
    }

//    public static void main(String[] args) {
//        try {
//          //  ZkListener zkListener = new ZkListener();
//            getCuratorFramework();
//
//
//        } catch (Exception e) {
//            System.out.println(ExceptionUtils.getStackTrace(e));
//        }
//    }
}
